package com.aiolos.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

/**
 *
 * 使用HDFS JAVA API实现词频统计————重构代码
 *
 * 需求：统计HDFS上的文件的wordCount，然后将统计结果输出到HDFS上
 *
 * 功能拆解：
 *    1) 读取HDFS上的文件  ==> HDFS API
 *    2) 业务处理(词频统计)：对文件中的每一行数据都要进行业务处理，按照分隔符分割    ==> Mapper
 *    3) 将处理结果缓存起来   ==> Context
 *    4) 将结果输出到HDFS上  ==> HDFS API
 */
public class HDFSWCApp02 {

    public static void main(String[] args) throws Exception{

        Properties properties = ConfigUtils.getProperties();

        //1) 读取HDFS上的文件
        //Path input = new Path("/hadoopapi/test/hello.txt");
        Path input = new Path(properties.getProperty(Constants.INPUT_PATH));

        // 获取到要操作的HDFS文件系统
        //URI uri = new URI("hdfs://192.168.126.129:8020");
        URI uri = new URI(properties.getProperty(Constants.HDFS_URI));
        Configuration conf = new Configuration();
        //设置默认副本系数为1
        conf.set("dfs.replication",properties.getProperty(Constants.DFS_REPLICATION));
        String userName = properties.getProperty(Constants.HADOOP_USER_NAME);
        FileSystem fileSystem = FileSystem.get(uri,conf,userName);
        // 获取HDFS下文件夹下的所有文件
        RemoteIterator<LocatedFileStatus> iterator = fileSystem.listFiles(input,false);
        // 实例化mapper实现类
        //ImoocMapper mapper = new WordCountMapperImpl();

        //通过反射实例化mapper的实现类
        Class<?> clazz = Class.forName(properties.getProperty(Constants.MAPPER_CLASS));
        ImoocMapper mapper = (ImoocMapper)clazz.newInstance();

        // 实例化上下文
        ImoocContext context = new ImoocContext();
        while (iterator.hasNext()){
            LocatedFileStatus file = iterator.next();
            FSDataInputStream in = fileSystem.open(file.getPath());
            BufferedReader reader = new BufferedReader(new InputStreamReader(in));

            String line = "";
            while ((line = reader.readLine()) != null){
                //2. 业务处理(词频统计) 返回结果：（hello，3）
                //在业务逻辑完成之后将结果写入到上下文中去
                mapper.map(line,context);
            }

            reader.close();
            in.close();
        }


        //3) 将处理结果缓存到上下文  Map
        Map<Object,Object> contextMap = context.getCacheMap();
        //4) 将结果输出到HDFS上
        Path output = new Path(properties.getProperty(Constants.OUTPUT_PATH));
        FSDataOutputStream out = fileSystem.create(new Path(output,properties.getProperty(Constants.OUTPUT_FILE)));

        //  将第三步缓存中的内容输出到out中
        Set<Map.Entry<Object,Object>> entries = contextMap.entrySet();
        for(Map.Entry<Object,Object> entry:entries){
            out.writeUTF(entry.getKey().toString()+" \t " + entry.getValue().toString() + "\n");
        }

        out.close();
        fileSystem.close();

        System.out.println("PK哥的HDFS API统计框架V2.0运行完成");
    }


}
